#pragma once
#include <queue>
#include <pthread.h>
#include "Task.hpp"
class Threadpool
{
private:
    Threadpool(int threadNum);
    Threadpool(const Threadpool &) = delete;
    Threadpool &operator=(const Threadpool &) = delete;
    ~Threadpool();

public:
    // 获得线程池单例
    static Threadpool *GetInstance(int threadNum)
    {
        // Double Check Lock
        if (instance == nullptr)
        {
            pthread_mutex_lock(&Mutex);
            if (instance == nullptr)
            {
                instance = new Threadpool(threadNum);
            }
            pthread_mutex_unlock(&Mutex);
        }
        return instance;
    }
    // 启动线程池
    void loop()
    {
        for (int i = 0; i < threadNum_; i++)
        {
            pthread_create(&threads[i], nullptr, run, nullptr);
        }
    }
    // 添加任务到任务队列
    void addTask(void (*func)(void *), void *args);
    // 添加空任务到任务队列
    void joinAll();

private:
    static void *run(void *args);

    static Threadpool *instance;
    std::queue<Task *> Taskqueue; // 任务队列
    pthread_mutex_t queueMutex;   // 锁
    static pthread_mutex_t Mutex; // 单例锁
    pthread_cond_t queueCond;     // 条件变量
    pthread_t *threads;
    int threadNum_; // 线程数量
    bool stop;
};
Threadpool *Threadpool::instance = nullptr;
pthread_mutex_t Threadpool::Mutex;

void Threadpool::addTask(void (*func)(void*), void *args)
{
    Task *task = new Task(func, args);
    pthread_mutex_lock(&queueMutex);
    Taskqueue.push(task);
    // 通知消费者消费
    pthread_cond_signal(&queueCond);
    pthread_mutex_unlock(&queueMutex);
}

Threadpool::~Threadpool()
{
    stop = true;
    // 唤醒所有线程
    pthread_cond_broadcast(&queueCond);
    for (int i = 0; i < threadNum_; i++)
    {
        pthread_join(threads[i], nullptr);
    }
    delete[] threads;
    pthread_mutex_destroy(&queueMutex);
    pthread_cond_destroy(&queueCond);
}

void *Threadpool::run(void *args)
{
    while (true)
    {
        pthread_mutex_lock(&instance->queueMutex);
        while (instance->Taskqueue.empty() && !instance->stop)
        {
            // 等待线程被唤醒
            pthread_cond_wait(&instance->queueCond, &instance->queueMutex);
        }
        if (instance->stop)
        {
            pthread_mutex_unlock(&instance->queueMutex);
            pthread_exit(nullptr);
        }
        Task *task = instance->Taskqueue.front();
        instance->Taskqueue.pop();
        pthread_mutex_unlock(&instance->queueMutex);
        if (task)
        {
            task->func_(task->args_);
            delete task;
        }
    }
    return nullptr;
}

// 向队列添加空任务
void Threadpool::joinAll()
{
    for (int i = 0; i < threadNum_; i++)
    {
        addTask(nullptr, nullptr);
    }
}

Threadpool::Threadpool(int threadNum)
{
    this->threadNum_ = threadNum;
    pthread_mutex_init(&queueMutex, nullptr);
    pthread_cond_init(&queueCond, nullptr);
    threads = new pthread_t[threadNum];
    stop = false;
}
